package com.atguigu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.util.Random

/**
  * create by undeRdoG on  2021-06-19  18:02
  * 凡心所向，素履以往，生如逆旅，一苇以航。
  */
object SparkStreaming05_State {

  def main(args: Array[String]): Unit = {
    // 创建环境
    /**
      * StreamingContext  创建时需要两个参数
      * 第一个参数环境配置
      * 第二个参数 批处理周期  duration
      **/
    val sparkConf = new SparkConf().setAppName("Streaming").setMaster("local[*]")
    val ssc = new StreamingContext(sparkConf, Seconds(3))


    // 获取端口数据
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 9999)

    val words: DStream[String] = lines.flatMap(_.split(" "))


    val wordToOne: DStream[(String, Int)] = words.map(x => (x, 1))

    //val res: DStream[(String, Int)] = wordToOne.reduceByKey(_ + _)

    /**
      * updateStateByKey :  根据 key 对数据的状态进行更新
      * 传递的参数中有两个值，
      * 第一个值表示  本次采集周期相同的 key的 value数据
      * 第二个值表示  缓冲区中相同key的  value 数据
      *
      * 保存状态必须  显式的 设置 checkpoint 的 存储路径，
      * 缓冲区中的数据，就是存在 checkpoint 中
      **/

      ssc.checkpoint("/cp`")

    val res: DStream[(String, Int)] = wordToOne.updateStateByKey(
      (seq: Seq[Int], buff: Option[Int]) => {
        val newCount = buff.getOrElse(0) + seq.sum
        // 将新数据放在缓冲区中
        Option(newCount)
      }
    )
    res
    res.print()


    /**
      * 由于sparkStreaming 采集器是长期执行的任务，所以不能直接关闭
      * 如果main方法执行完毕，程序也会结束
      * 因此不能让main方法执行完毕
      **/

    // 启动采集器
    ssc.start()

    // 等待采集器关闭
    ssc.awaitTermination()

  }

}
